package fun.taostool.service.taost.impl;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import fun.taostool.config.ProjectConfig;
import fun.taostool.entity.sys.EUnitEntity;
import fun.taostool.entity.taost.TaostEntity;
import fun.taostool.mapper.mysql.DeviceInfoMapper;
import fun.taostool.mapper.mysql.TaostMapper;
import fun.taostool.service.sys.EUnitService;
import fun.taostool.service.taost.TaostService;
import fun.taostool.vo.DeviceInfoVO;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.avro.Schema;
import org.apache.avro.file.DataFileReader;
import org.apache.avro.file.DataFileWriter;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.specific.SpecificDatumWriter;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @description: TestServiceImpl
 * @author: Chensn
 * @time: 2024/11/26
 */
@Service
@Slf4j
public class TaostServiceImpl extends ServiceImpl<TaostMapper, TaostEntity> implements TaostService {
    @Resource
    @Qualifier("taosJdbcTemplate")
    private JdbcTemplate jdbcTemplate;
    // 是否中断备份/恢复程序
    private volatile boolean isInterrupted = false;
    private static final String NODATA = "nodata";

    // 数据恢复插入批次大小
    @Value("${batch.size}")
    private int RECOVERY_BATCH_SIZE;
    @Resource
    @Qualifier("taskExecutor")
    private ThreadPoolTaskExecutor pool;

    @Resource
    private EUnitService eUnitService;

    @Resource
    private DeviceInfoMapper deviceInfoMapper;
    @Resource
    private ProjectConfig projectConfig;

    /**
     * 数据备份
     *
     * @param tableName 备份表
     */
    @Override
    public void dataBackup(String tableName) {
        // 每次查询计划100条
        int batchSize = 100;
        int timeStepConfig;
        if (projectConfig.getTimeStep() <= 0) {
            timeStepConfig = 10;
        } else {
            timeStepConfig = projectConfig.getTimeStep();
        }
        // 时间步长 10天
        long timeStep = 86400000L * timeStepConfig;
        log.info("开始备份,步长：" + timeStepConfig);
        String schemaJson = "{\"type\":\"record\",\"name\":\"_record\",\"namespace\":\"tag_st\",\"fields\":[{\"name\":\"ts\",\"type\":[\"null\",\"long\"]},{\"name\":\"value\",\"type\":[\"null\",\"float\"]},{\"name\":\"aste\",\"type\":[\"null\",\"float\"]},{\"name\":\"qul\",\"type\":[\"null\",\"float\"]}]}";
        // 定义要创建的文件夹和文件路径
        CompletableFuture.runAsync(() -> {
            LambdaQueryWrapper<TaostEntity> countQW = new LambdaQueryWrapper<>();
            countQW.eq(TaostEntity::getType, 1)
                    .eq(StringUtils.isNotBlank(tableName), TaostEntity::getTbname, tableName)
                    .eq(TaostEntity::getState, 0);
            int count = this.count(countQW);
            // 计算总批次，向上取整
//            int totalBatches = (count + batchSize - 1) / batchSize;
            AtomicInteger sp = new AtomicInteger();

//            for (int batch = 0; batch < totalBatches; batch++) {
            // 计算偏移量
//                int mysqlOffset = batch * batchSize;
            LambdaQueryWrapper<TaostEntity> queryWrapper = new LambdaQueryWrapper<>();
            queryWrapper.eq(TaostEntity::getType, 1)
                    .eq(TaostEntity::getState, 0)
                    .eq(StringUtils.isNotBlank(tableName), TaostEntity::getTbname, tableName);

            // 使用偏移量和限制来查询数据
//                List<TaostEntity> batchData = this.list(queryWrapper.last("LIMIT " + batchSize + " OFFSET " + mysqlOffset + ";"));
            List<TaostEntity> batchData = this.list(queryWrapper);

            List<Future<?>> futures = new ArrayList<>();
            for (TaostEntity batchDatum : batchData) {
                futures.add(pool.submit(() -> processTaost(batchDatum, timeStep, schemaJson, sp, count)));
            }
            futures.removeIf(Future::isDone);
//            }
        }, pool);


    }

    private void processTaost(TaostEntity batchDatum, long timeStep, String schemaJson, AtomicInteger sp, int count) {
        long startTimeOne = System.currentTimeMillis();
        log.info("正在备份：" + batchDatum.getTbname());
        // 每张普通表一个存储路径
        File filepath = new File("data/" + batchDatum.getEuid() + "/" + batchDatum.getLocationCode());
        if (!filepath.exists()) {
            if (!filepath.mkdirs()) {
                log.error("创建文件存储目录失败！data/" + batchDatum.getEuid() + "/" + batchDatum.getLocationCode());
                throw new RuntimeException();
            }
        }
        // avro 不支持文件续写 只能一次性写入 一次性读取
        File file = new File("data/" + batchDatum.getEuid() + "/" + batchDatum.getLocationCode(), batchDatum.getTbname() + ".avro");
        Schema schema = new Schema.Parser().parse(schemaJson);
        GenericRecord record = new GenericData.Record(schema);
        boolean isEnd = true;

        long bkCount = 0;
        long starTimes = batchDatum.getStartTime().getTime();
        long endTimes = batchDatum.getEndTime().getTime();

        long breakTimes = starTimes + timeStep;
        try (DataFileWriter<GenericRecord> dataFileWriter = new DataFileWriter<>(new SpecificDatumWriter<>(schema))) {
            dataFileWriter.create(schema, file);
            while (isEnd) {

                if (breakTimes >= endTimes) {
                    breakTimes = endTimes;
                    isEnd = false;
                }
                if (isInterrupted) {
                    batchDatum.setState(0);
                    batchDatum.setBreakTime(null);
                    this.updateById(batchDatum);
                    return;
                }
                StringBuffer sqlQuery = new StringBuffer();
                sqlQuery.append("SELECT TBNAME ,* FROM device_point.tag_st WHERE TBNAME = '")
                        .append(batchDatum.getTbname())
                        .append("' and ts >= ")
                        .append(starTimes)
                        .append(" and ts < ")
                        .append(breakTimes)
                        .append(";");
                List<Map<String, Object>> resultList = jdbcTemplate.queryForList(sqlQuery.toString());
                for (Map<String, Object> result : resultList) {
                    bkCount++;
                    Date tsTime = (Date) result.get("ts");
                    long time = tsTime.getTime();
                    // 根据查询结果的列名和数据类型填充Avro记录
                    record.put("ts", time);
                    record.put("value", result.get("value"));
                    record.put("aste", result.get("aste"));
                    record.put("qul", result.get("qul"));
                    dataFileWriter.append(record);
                }
                starTimes = breakTimes;
                breakTimes += timeStep;
            }
        } catch (Exception e) {
            log.error("备份异常: " + e.getMessage());
            throw new RuntimeException(e);
        }
        sp.getAndIncrement();
        if (bkCount == 0) {
            Path path = Paths.get(file.getAbsolutePath());
            try {
                Files.delete(path);
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
            // 无需备份
            batchDatum.setState(3);
            batchDatum.setBkpath(NODATA);
            this.updateById(batchDatum);
        } else {
            double fileSizeInMB = file.length() / (1024.0 * 1024.0);
            String formattedSize = String.format("%.2f MB", fileSizeInMB);
            log.info("备份已完成,文件大小:" + formattedSize + "条数" + bkCount);
            batchDatum.setBkpath(file.getAbsolutePath());
            batchDatum.setFinishTime(new Date());
            batchDatum.setState(3);
            batchDatum.setBreakTime(new Date(breakTimes - timeStep));
            this.updateById(batchDatum);
        }
        // /57/4613
        long endTimeOne = System.currentTimeMillis();
        log.info("备份用时: " + ((endTimeOne - startTimeOne) / 1000) + " 秒");
        log.info("备份进度：" + sp + "/" + count);
    }

    @Override
    public boolean interruptedProgram(boolean interrupted) {
        this.isInterrupted = true;
        return isInterrupted;

    }


    /**
     * 生成备份计划
     *
     * @return
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    @SneakyThrows
    public boolean bkPlanGenerate() {
        List<TaostEntity> taostEntityList = new ArrayList<>();
        List<EUnitEntity> list = eUnitService.list();
        SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd");
        List<String> types = Arrays.asList("BMM", "WHM", "PCS(a)", "PCS(m)", "BMS", "BMC", "EU", "DU");
        List<String> bmmTags = new ArrayList<>(Arrays.asList(
                "Conn", "Online", "P", "Q", "S", "F", "PF", "EPEpk", "EPEpkt", "EEpct",
                "EPEvlt", "EPIpk", "EPIpkt", "EPIpct", "EPIvlt", "EPITot", "EPETot",
                "PAPdemax", "RAPdemax", "PAPdemaxTime", "RAPdemaxTime", "Pdemand",
                "EPI", "EPILM", "SafeValue"
        ));

        List<String> whmTags = new ArrayList<>(Arrays.asList(
                "Online", "P", "Q", "S", "F", "PF", "EPI", "EPE", "Pdemax"
        ));

        List<String> pcsTags = new ArrayList<>(Arrays.asList(
                "RP", "F", "P", "Q", "S", "PF", "SumChgE", "SumDsgE", "DayChgE",
                "DayDsgE", "Pset", "Qset", "SysON", "SysOFF", "CtlType", "Order",
                "CtlSta"
        ));

        List<String> bmsTags = new ArrayList<>(Arrays.asList(
                "SysTotU", "SysTotI", "SysSOC", "SysSOH", "SysIns", "SysChgbleImax",
                "SysDsgbleImax", "SysUmax", "SysUmin", "SysUavg", "SysTmax", "SysTmin",
                "SysTavg", "DayChgE", "DayDsgE", "SumChgE", "SumDsgE", "DsgblePmax",
                "ChgblePmax", "SysChgbleE", "SysDsgbleE"
        ));

        List<String> euTags = new ArrayList<>(Arrays.asList(
                "UserLoadP", "DayChgE", "DayDsgE", "SumChgE", "SumDsgE", "LastDayChgE",
                "LastDayDsgE", "ULPdeKw", "ULPdemaxKw", "PLDistance", "ReferenceMaxDemand",
                "DefaultMaxDemand", "ESPdemaxKw", "ESPdeKw", "TPdeKw", "TPdemaxKw",
                "SysESoc"
        ));

        List<String> duTags = new ArrayList<>(Arrays.asList(
                "CPU", "Memory", "DiskMax", "DiskAvg", "NetState", "NetAbnormalTime",
                "Goroutine", "Conn"
        ));


        Date now = new Date();
        for (EUnitEntity eUnitEntity : list) {
            String euid = eUnitEntity.getLocationCode();
            String lowerCaseEuid = euid.toLowerCase();
            String enableDate = eUnitEntity.getEnableDate();
            if (StringUtils.isEmpty(enableDate)) {
                continue;
            }
            // 定义日期格式
            Date date = formatter.parse(enableDate);
            for (String type : types) {
                List<DeviceInfoVO> codeList = new ArrayList<>();
                if (type.equals("DU")) {
                    DeviceInfoVO deviceInfoVO = new DeviceInfoVO();
                    deviceInfoVO.setLocationCode(euid + "-DU1");
                    deviceInfoVO.setEContId(null);
                    codeList.add(deviceInfoVO);
                } else if (type.equals("EU")) {
                    DeviceInfoVO deviceInfoVO = new DeviceInfoVO();
                    deviceInfoVO.setLocationCode(euid);
                    deviceInfoVO.setEContId(null);
                    codeList.add(deviceInfoVO);
                } else {
                    codeList = deviceInfoMapper.getLocationCodeByEU(eUnitEntity.getId(), type);
                }
                for (DeviceInfoVO vo : codeList) {
                    switch (type) {
                        case "BMM": {
                            for (String bmmTag : bmmTags) {
                                TaostEntity taostEntity = new TaostEntity();
                                taostEntity.setEuid(lowerCaseEuid)
                                        .setLocationCode(vo.getLocationCode())
                                        .setTbname(vo.getLocationCode() + "_" + bmmTag)
                                        .setState(0)
                                        .setTag(bmmTag)
                                        .setType(1)
                                        .setEcid(vo.getEContId())
                                        .setStartTime(date)
                                        .setEndTime(now);
                                taostEntityList.add(taostEntity);
                            }
                            break;
                        }
                        case "WHM": {
                            for (String whmTag : whmTags) {
                                TaostEntity taostEntity = new TaostEntity();
                                taostEntity.setEuid(lowerCaseEuid)
                                        .setLocationCode(vo.getLocationCode())
                                        .setTbname(vo.getLocationCode() + "_" + whmTag)
                                        .setState(0)
                                        .setTag(whmTag)
                                        .setEcid(vo.getEContId())
                                        .setType(1)
                                        .setStartTime(date)
                                        .setEndTime(now);
                                taostEntityList.add(taostEntity);
                            }
                            break;
                        }
                        case "PCS(a)": {
                            for (String pcsTag : pcsTags) {
                                TaostEntity taostEntity = new TaostEntity();
                                taostEntity.setEuid(lowerCaseEuid)
                                        .setLocationCode(vo.getLocationCode())
                                        .setTbname(vo.getLocationCode() + "_" + pcsTag)
                                        .setEcid(vo.getEContId())
                                        .setState(0)
                                        .setTag(pcsTag)
                                        .setType(1)
                                        .setStartTime(date)
                                        .setEndTime(now);
                                taostEntityList.add(taostEntity);
                            }
                            break;
                        }
                        case "PCS(m)": {
                            for (String pcsTag : pcsTags) {
                                TaostEntity taostEntity = new TaostEntity();
                                taostEntity.setEuid(lowerCaseEuid)
                                        .setLocationCode(vo.getLocationCode())
                                        .setEcid(vo.getEContId())
                                        .setTbname(vo.getLocationCode() + "_" + pcsTag)
                                        .setTag(pcsTag)
                                        .setState(0)
                                        .setStartTime(date)
                                        .setType(1)
                                        .setEndTime(now);
                                taostEntityList.add(taostEntity);
                            }
                            break;
                        }
                        case "BMS": {
                            for (String bmsTag : bmsTags) {
                                TaostEntity taostEntity = new TaostEntity();
                                taostEntity.setEuid(lowerCaseEuid)
                                        .setLocationCode(vo.getLocationCode())
                                        .setEcid(vo.getEContId())
                                        .setTbname(vo.getLocationCode() + "_" + bmsTag)
                                        .setState(0)
                                        .setType(1)
                                        .setTag(bmsTag)
                                        .setStartTime(date)
                                        .setEndTime(now);
                                taostEntityList.add(taostEntity);
                            }
                            break;
                        }
                        case "BMC": {
                            for (String bmsTag : bmsTags) {
                                TaostEntity taostEntity = new TaostEntity();
                                taostEntity.setEuid(lowerCaseEuid)
                                        .setLocationCode(vo.getLocationCode())
                                        .setEcid(vo.getEContId())
                                        .setType(1)
                                        .setTbname(vo.getLocationCode() + "_" + bmsTag)
                                        .setState(0)
                                        .setTag(bmsTag)
                                        .setStartTime(date)
                                        .setEndTime(now);
                                taostEntityList.add(taostEntity);
                            }
                            break;
                        }
                        case "EU": {
                            for (String euTag : euTags) {
                                TaostEntity taostEntity = new TaostEntity();
                                taostEntity.setEuid(lowerCaseEuid)
                                        .setLocationCode(vo.getLocationCode())
                                        .setTbname(vo.getLocationCode() + "_" + euTag)
                                        .setState(0)
                                        .setStartTime(date)
                                        .setEcid(vo.getEContId())
                                        .setTag(euTag)
                                        .setType(1)
                                        .setEndTime(now);
                                taostEntityList.add(taostEntity);
                            }
                            break;
                        }
                        case "DU": {
                            for (String duTag : duTags) {
                                TaostEntity taostEntity = new TaostEntity();
                                taostEntity.setEuid(lowerCaseEuid)
                                        .setLocationCode(vo.getLocationCode())
                                        .setType(1)
                                        .setTbname(vo.getLocationCode() + "_" + duTag)
                                        .setState(0)
                                        .setTag(duTag)
                                        .setStartTime(date)
                                        .setEcid(null)
                                        .setEndTime(now);
                                taostEntityList.add(taostEntity);
                            }
                            break;
                        }
                        default:
                            break;
                    }
                }
            }
        }
        return this.saveBatch(taostEntityList);
    }

    /**
     * 生成恢复计划
     *
     * @return
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public boolean rePlanGenerate() {
        LambdaQueryWrapper<TaostEntity> qw = new LambdaQueryWrapper<>();
        qw.ne(TaostEntity::getBkpath, NODATA)
                .isNotNull(TaostEntity::getBkpath)
                .eq(TaostEntity::getType, 1)
                .eq(TaostEntity::getState, 3);
        List<TaostEntity> list = this.list(qw);
        List<TaostEntity> reList = new ArrayList<>(list);
        for (TaostEntity taostEntity : reList) {
            taostEntity.setType(2);
            taostEntity.setId(null);
            taostEntity.setState(0);
            taostEntity.setFinishTime(null);
            taostEntity.setBreakTime(null);
        }
        boolean ok = this.saveBatch(reList);
        if (ok) {
            log.info("生成恢复计划条数：" + reList.size());
        }
        return ok;
    }


    /**
     * 数据恢复
     *
     * @param tableName 普通表名
     * @return
     */
    @Override
    public boolean dataRecovery(String tableName) {

        CompletableFuture.runAsync(() -> {
            int batchSize = 10000;
            LambdaQueryWrapper<TaostEntity> countQW = new LambdaQueryWrapper<>();
            countQW.eq(TaostEntity::getType, 2)
                    .eq(StringUtils.isNotBlank(tableName), TaostEntity::getTbname, tableName)
                    .eq(TaostEntity::getState, 0);
            int count = this.count(countQW);
            int thisCount = 0;
            log.info("数据恢复开始,预计恢复普通表数量：" + count);
            log.info("taos每次插入批次大小：" + RECOVERY_BATCH_SIZE);

            // 计算总批次，向上取整
//            int totalBatches = (count + batchSize - 1) / batchSize;

//            for (int batch = 0; batch < totalBatches; batch++) {
            // 计算偏移量
//                int mysqlOffset = batch * batchSize;
            LambdaQueryWrapper<TaostEntity> taostEntityQW = new LambdaQueryWrapper<>();
            taostEntityQW.eq(TaostEntity::getType, 2)
                    .eq(TaostEntity::getState, 0)
                    .eq(StringUtils.isNotBlank(tableName), TaostEntity::getTbname, tableName);
            // 使用偏移量和限制来查询数据
//                List<TaostEntity> batchData = this.list(taostEntityQW.last("LIMIT " + batchSize + " OFFSET " + mysqlOffset + ";"));
            List<TaostEntity> batchData = this.list(taostEntityQW);
            for (TaostEntity batchDatum : batchData) {
                long startTime = System.currentTimeMillis();
                thisCount++;
                int reCount = 0;
                if (isInterrupted) {
                    batchDatum.setState(0);
                    batchDatum.setBreakTime(null);
                    this.updateById(batchDatum);
                    return;
                }
                String euid = batchDatum.getEuid();
                String euidS = euid.replaceAll("[a-zA-Z]", "").replaceAll("\\(", "").replaceAll("\\)", "");
                int euIdNumber = Integer.parseInt(euidS);

                // 从Avro文件恢复数据并导入TDengine
                try (DataFileReader<GenericRecord> reader = new DataFileReader<>(new File(batchDatum.getBkpath()), new GenericDatumReader<>())) {
                    List<GenericRecord> insertBatch = new ArrayList<>(RECOVERY_BATCH_SIZE);
                    GenericRecord record;
                    while (reader.hasNext()) {
                        record = reader.next(null);
                        insertBatch.add(record);
                        if (insertBatch.size() == RECOVERY_BATCH_SIZE) {
                            reCount += processBatch(insertBatch, euid, euIdNumber, batchDatum);
                            // 清空批次以便下一批数据
                            insertBatch.clear();
                        }
                    }
                    // 处理最后一批（可能小于BATCH_SIZE）
                    if (!insertBatch.isEmpty()) {
                        reCount += processBatch(insertBatch, euid, euIdNumber, batchDatum);
                    }

                } catch (IOException e) {
                    throw new RuntimeException(e);
                }

                batchDatum.setFinishTime(new Date());
                batchDatum.setState(3);
                this.updateById(batchDatum);
                log.info(batchDatum.getTbname() + " 数据已恢复条数：" + reCount);
                long endTime = System.currentTimeMillis();
                log.info("恢复用时：" + ((endTime - startTime) / 1000) + "秒");
                log.info("恢复进度：" + thisCount + "/" + count);
            }
//            }
        }, pool);
        return true;
    }

    /**
     * 批次插入处理
     *
     * @param batch
     * @param euid
     * @param euIdNumber
     * @param batchDatum
     * @return
     */
    private int processBatch(List<GenericRecord> batch, String euid, Integer euIdNumber, TaostEntity batchDatum) {
        int thisCount = 0;
        StringBuffer insertSql = new StringBuffer("insert into ");
        if (batchDatum.getEcid() != null) {
            insertSql.append(euid + "_device_point.`" + batchDatum.getTbname() + "` using ")
                    .append(euid + "_device_point.`tag_st` tags ")
                    .append("(" + euIdNumber + "," + batchDatum.getEcid() + ",'" + batchDatum.getLocationCode() + "','" + batchDatum.getTag() + "')")
                    .append(" (`ts`,`value`,`aste`,`qul`) values ");
        } else {
            insertSql.append(euid + "_device_point.`" + batchDatum.getTbname() + "` using ")
                    .append(euid + "_device_point.`tag_st` (`euId`,`locationCode`,`tagName`) tags ")
                    .append("(" + euIdNumber + ",'" + batchDatum.getLocationCode() + "','" + batchDatum.getTag() + "')")
                    .append(" (`ts`,`value`,`aste`,`qul`) values ");
        }
        for (GenericRecord record : batch) {
            insertSql.append("(" + record.get("ts") + ",")
                    .append(record.get("value") + ",")
                    .append(record.get("aste") + ",")
                    .append(record.get("qul") + ")");
            thisCount++;
        }
        insertSql.append(";");
        jdbcTemplate.execute(insertSql.toString());
        return thisCount;
    }


    /**
     * avro读取测试
     *
     * @param avroFilePath avro文件路径
     * @return
     */
    @Override
    public String avroRead(String avroFilePath) {
        StringBuilder sb = new StringBuilder();
        // 读取Avro文件
        try (DataFileReader<GenericRecord> reader = new DataFileReader<>(new File(avroFilePath), new GenericDatumReader<>())) {
            Schema schema = reader.getSchema();
            System.out.println("Schema: " + schema);
            sb.append("Schema: " + schema);
            int count = 0;
            // 逐条读取记录
            for (GenericRecord record : reader) {
                count++;
                System.out.println(record);
//                sb.append("Record: " + record);
            }
            System.out.println(reader.getBlockSize());
            sb.append(" BlockSize:" + reader.getBlockSize());
            sb.append(" BlockCount:" + reader.getBlockCount());
            sb.append(" totalCount:" + count);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return sb.toString();
    }


//
//
//    @SneakyThrows
//    private void backupDataWriting(String dbName, String stbName, String euid, int pageSize, int offset, Date startTime, Date endTime) {
//        // 使用当前数据库
//        jdbcTemplate.execute("USE " + dbName + ";");
//        // Stable建表名称
//        String dbQuery = "SHOW CREATE TABLE " + stbName + ";";
//        SqlRowSet dbRowSet = jdbcTemplate.queryForRowSet(dbQuery);
//
//        String schemaJson = null;
//        while (dbRowSet.next()) {
//            String createSTable = dbRowSet.getString("Create Table");
//            // 解析建表语句并创建Avro schema JSON对象
//            JSONObject avroSchema = parseCreateTableStatement(createSTable);
//            JSONArray fields = avroSchema.getJSONArray("fields");
//            int length = fields.length();
//            System.out.println(length);
//            List<Field> fieldList = new ArrayList<>();
//            for (int i = 0; i < fields.length(); i++) {
//                JSONObject jsonObject = fields.getJSONObject(i);
//                String name = jsonObject.getString("name");
//
//                // 获取 type 数组并转换为 List<Object>
//                JSONArray typeArray = jsonObject.getJSONArray("type");
//                List<Object> typeList = new ArrayList<>();
//                for (int j = 0; j < typeArray.length(); j++) {
//                    typeList.add(typeArray.opt(j)); // 使用 opt 方法以避免 NullPointerException
//                }
//
//                Field field = new Field(name, typeList);
//                fieldList.add(field);
//            }
//            for (Field field : fieldList) {
//                String name = field.getName();
//                List<Object> type = field.getType();
//                System.out.println("name: " + name);
//                System.out.println("type0: " + type.get(0));
//                System.out.println("type1: " + type.get(1));
//
//            }
//            schemaJson = avroSchema.toString();
//            // 打印Avro schema JSON
//            System.out.println(createSTable);
//            System.out.println(avroSchema); // 格式化输出，缩进4个空格
//            if (schemaJson == null) {
//                log.error("解析建表sql失败：" + createSTable);
//                return;
//            }
//        }
//        File file = new File(dbName + "_" + stbName + ".avro");
//        Schema schema = new Schema.Parser().parse(schemaJson);
//        // 写入Avro文件
//        GenericRecord record = new GenericData.Record(schema);
//
//        boolean hasMoreData = true;
//        while (hasMoreData && !isInterrupted) {
//            try (DataFileWriter<GenericRecord> dataFileWriter = new DataFileWriter<>(new SpecificDatumWriter<>(schema))) {
//                dataFileWriter.create(schema, file);
//                StringBuffer sqlQuery = new StringBuffer();
//                sqlQuery.append("select tbname ,* from ")
//                        .append(dbName)
//                        .append(".")
//                        .append(stbName)
//                        .append(" where tbname like '")
//                        .append(euid)
//                        .append("%' and ts >= ")
//                        .append(startTime.getTime())
//                        .append(" and ts <= ")
//                        .append(endTime.getTime())
//                        .append("and ")
//                        .append(" limit ")
//                        .append(pageSize)
//                        .append(" OFFSET ")
//                        .append(offset)
//                        .append(";");
//                String finallySQl = sqlQuery.toString();
//                List<Map<String, Object>> resultList = jdbcTemplate.queryForList(finallySQl);
//                if (resultList.isEmpty()) {
//                    hasMoreData = false;
//                } else {
//                    Thread.sleep(100);
//                    offset += pageSize;
//                    for (Map<String, Object> result : resultList) {
//                        Date tsTime = (Date) result.get("ts");
//                        long time = tsTime.getTime();
//                        String tbname = result.get("tbname").toString();
//                        String[] tbParts = tbname.split("[-_]");
//                        String deviceType;
//                        if (tbParts.length >= 3) {
//                            String part = tbParts[tbParts.length - 2];
//                            deviceType = part;
//                        } else if (tbParts.length == 2) {
//                            String part = tbParts[0];
//                            deviceType = part;
//                            System.out.println(part);
//                        } else {
//                            continue;
//                        }
//                        System.out.println(deviceType);
//
//                        // 根据查询结果的列名和数据类型填充Avro记录
//                        record.put("tbname", result.get("tbname"));
//                        record.put("ts", time);
//                        record.put("value", result.get("value"));
//                        record.put("aste", result.get("aste"));
//                        record.put("qul", result.get("qul"));
//                        record.put("euId", result.get("euId"));
//                        record.put("ecId", result.get("ecId"));
//                        record.put("locationCode", result.get("locationCode"));
//                        record.put("tagName", result.get("tagName"));
//                        dataFileWriter.append(record);
//                    }
//                }
//            } catch (Exception e) {
//                log.error("备份异常: " + e.getMessage());
//                throw new RuntimeException(e);
//            }
//        }
//        log.info("备份文件路径: " + file.getAbsolutePath());
//    }


//    /**
//     * 解析建表语句并返回Avro schema JSON对象的方法
//     *
//     * @param createTableStatement
//     * @return
//     */
//    @SneakyThrows
//    private static JSONObject parseCreateTableStatement(String createTableStatement) {
//        // 提取表名
//        String tableName = extractTableName(createTableStatement);
//
//        // 提取常规字段和TAG字段的字符串
//        String regularFieldsStr = extractFieldsStr(createTableStatement, "CREATE STABLE `" + tableName + "` (", ")");
//        String tagsFieldsStr = extractFieldsStr(createTableStatement, ") TAGS (", ")");
//
//        // 解析字段字符串，并合并到同一个JSONArray中
//        JSONArray allFields = new JSONArray();
//        addFieldsToArray(allFields, regularFieldsStr);
//        addFieldsToArray(allFields, tagsFieldsStr);
//
//        // 创建Avro schema JSON对象
//        JSONObject avroSchema = new JSONObject();
//        avroSchema.put("type", "record");
//        avroSchema.put("name", tableName);
//        avroSchema.put("fields", createAvroFields(allFields));
//
//        return avroSchema;
//    }
//
//    // 提取表名
//    private static String extractTableName(String createTableStatement) {
//        int startIndex = createTableStatement.indexOf("`") + 1;
//        int endIndex = createTableStatement.indexOf("`", startIndex);
//        return createTableStatement.substring(startIndex, endIndex);
//    }
//
//    // 提取字段字符串
//    private static String extractFieldsStr(String createTableStatement, String startDelimiter, String endDelimiter) {
//        int startIndex = createTableStatement.indexOf(startDelimiter) + startDelimiter.length();
//        int endIndex = createTableStatement.indexOf(endDelimiter, startIndex);
//        if (endIndex == -1) {
//            endIndex = createTableStatement.length();
//        }
//        return createTableStatement.substring(startIndex, endIndex).trim();
//    }

    // 解析字段字符串，并将字段添加到JSONArray中的方法
//    @SneakyThrows
//    private static JSONArray addFieldsToArray(JSONArray fields, String fieldsStr) {
//        if (!fieldsStr.isEmpty()) {
//            String[] fieldsArray = fieldsStr.split(",\\s*");
//            for (String field : fieldsArray) {
//                field = field.trim();
//                int nameEndIndex = field.indexOf(" ");
//                if (nameEndIndex != -1) {
//                    String name = field.substring(0, nameEndIndex).replace("`", "");
//                    String type = field.substring(nameEndIndex + 1).trim().replaceAll("[`()]", "");
//
//                    // 根据需要转换类型
//                    type = convertDatabaseTypeToAvroType(type);
//
//                    JSONObject fieldObject = new JSONObject();
//                    fieldObject.put("name", name);
//                    fieldObject.put("type", type);
//                    fields.put(fieldObject);
//                }
//            }
//        }
//        return fields;
//    }
//
//    // 根据字段数组创建Avro字段数组的方法
//    @SneakyThrows
//    private static JSONArray createAvroFields(JSONArray fields) {
//        JSONArray avroFields = new JSONArray();
//        for (int i = 0; i < fields.length(); i++) {
//            JSONObject field = fields.getJSONObject(i);
//            JSONObject avroField = new JSONObject();
//            avroField.put("name", field.getString("name"));
//            JSONArray typeArray = new JSONArray();
//            typeArray.put(field.getString("type"));
//            typeArray.put("null");
//            avroField.put("type", typeArray);
//            avroFields.put(avroField);
//        }
//        JSONObject tbField = new JSONObject();
//        tbField.put("name", "tbname");
//        JSONArray newTypeArray = new JSONArray();
//        newTypeArray.put("null");
//        newTypeArray.put("string");
//        tbField.put("type", newTypeArray);
//        avroFields.put(tbField);
//
//        return avroFields;
//    }
//
//    // 将数据库类型转换为Avro类型的方法
//    private static String convertDatabaseTypeToAvroType(String dbType) {
//        switch (dbType.toUpperCase()) {
//            case "TIMESTAMP":
//                return "long";
//            case "FLOAT":
//                return "float";
//            case "INT":
//                return "int";
//            default:
//                return "string";
//        }
//    }

}

